package com.huan.hadoop.mr;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;

/**
 * WordCount 的 Mapper 类，对应着 MapTask
 * <p>
 * KEYIN: 表示map阶段输入kv中的k类型，在默认读组件下是 起始位置偏移量，因此是LongWritable
 * VALUEIN: 表示map阶段输入kv中的v类型,在默认读组件下是 读取一行内容，因此是 Text
 * KEYOUT: 标识map阶段输出kv中的k类型， 和实际的业务相关，在本例子中是输出的是单词，因此是Text
 * VALUEOUT: 标识map阶段输出kv中的v类型， 和实际的业务相关，在本例子中是输出的是单词次数1，因此是LongWritable
 * <p>
 * 默认的读组件是： TextInputFormat，该组件的读取行为是，一行一行的读取数据，返回kv键值对格式，
 * k:表示每一行的起始位置偏移量(注意每行的结束回车换行), v:表示这一行的内容
 *
 * @author huan.fu
 * @date 2023/7/8 - 11:12
 */
public class WordCountMapper extends Mapper<LongWritable, Text, Text, LongWritable> {

    private Text wordKey = new Text();
    private final LongWritable one = new LongWritable(1);

    /**
     * map任务开始前执行一次，只会执行一次
     */
    @Override
    protected void setup(Mapper<LongWritable, Text, Text, LongWritable>.Context context) throws IOException, InterruptedException {
        // ignore
    }

    /**
     * 每当TextInputFormat读取返回一个kv键值对，就调用一次map方法进行业务处理。
     * 默认情况下： map方法是基于行来处理数据
     */
    @Override
    protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, LongWritable>.Context context) throws IOException, InterruptedException {
        // 获取一行数据
        String line = value.toString();
        // 根据分隔符进行切割
        String[] words = line.split("\\s+");
        // 遍历数据
        for (String word : words) {
            // 输出数据，把每个单词标记为1
            wordKey.set(word);
            // 使用上下文对象，将数据输出
            context.write(wordKey, one);
        }
    }

    /**
     * map任务结束后执行一次，只会执行一次
     */
    @Override
    protected void cleanup(Mapper<LongWritable, Text, Text, LongWritable>.Context context) throws IOException, InterruptedException {
        // ignore
    }
}
